/*
 * Copyright 2012 The LoadTestAnalyzer Project
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 */

package org.yantsu.loadtestanalyzer.ltaimporter.importers;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.text.NumberFormat;
import java.text.ParseException;
import java.util.Locale;
import java.util.StringTokenizer;

public class Sar implements Importer {
	private String testName;
	private BufferedReader reader;
	private int lineNo;
	private Data data;
	private Data nextData;
	public String cat;
	private boolean endOfData = false;
	
	@Override
	public String getMapping() {
		return "{\"data_Sar\" : {" +
						"\"_parent\" : {\"type\" : \"series\"}," +
						"\"properties\" : {" +
							"\"timestamp\" : {\"type\" : \"date\"}," +
							"\"category\" : {\"type\" : \"string\", \"index\" : \"not_analyzed\"}," +
							"\"hostname\" : {\"type\" : \"string\", \"index\" : \"not_analyzed\"}" +
						"}" +
					"}" +
				"}";
	}

	@Override
	public String getTestname() {
		return testName;
	}

	@Override
	public Data next() {
		if(endOfData) {
			data = null;
		} else {
			if(nextData == null) {
				data = new Data();
			} else {
				data = nextData;
			}
		}
		
		try {
			String 	line = reader.readLine();
			if(line == null) { //no more data
				endOfData = true;
			} else { //fill in data
				while(line != null) {
					StringTokenizer tokenizer = new StringTokenizer(line, "\t");
					if(tokenizer.countTokens() != 6) {
						throw new IllegalStateException("Sar format error: Token count in line " + lineNo + " is not 6");
					}
					
					String hostname = tokenizer.nextToken();
					tokenizer.nextToken(); //interval
					String timestamp = tokenizer.nextToken();
					String category = tokenizer.nextToken();
					String type = tokenizer.nextToken();
					String value = tokenizer.nextToken();
					
					//validation and typing
					long timestampL = -1;
					try {
						timestampL = Long.valueOf(timestamp);
					} catch(NumberFormatException e) {
						throw new IllegalStateException("Timestamp retrieved in line "+ lineNo + " is not a number, was " + timestamp, e);
					}
					double valueD = 0.0;
					try {
						NumberFormat nf = NumberFormat.getInstance(Locale.GERMAN);
						valueD = nf.parse(value).doubleValue();
					} catch(NumberFormatException e) {
						throw new IllegalStateException("Value retrieved in line " + lineNo + " is not a number, was " + value, e);
					} catch (ParseException e) {
						throw new IllegalStateException("Value retrieved in line " + lineNo + " is not a number, was " + value, e);
					}
					timestampL = timestampL * 1000L;
	
					if(timestampL == -1) {
						throw new IllegalStateException("Timestamp in line " + lineNo + " is invalid, was " + timestamp);
					}
					
					if(data.timestamp != -1 && (data.timestamp != timestampL || !"-".equals(category) && !data.category.equals(category))) { //new group starting
						nextData = new Data();
						nextData.timestamp = timestampL;
						nextData.hostname = hostname; 
						nextData.values.put(type, valueD);
						if("-".equals(category)) {
							cat = type;
							nextData.category = cat;
						} else {
							cat = null;
							nextData.category = category;
						}
						
						break;
					} else {
						if(data.timestamp == -1) { //the very first data
							data.timestamp = timestampL;
							data.hostname = hostname;
							if("-".equals(category)) {
								cat = type;
								data.category = cat;
							} else {
								cat = null;
								data.category = category;
							}
						} else if(!data.hostname.equals(hostname)) {
							throw new IllegalStateException("Hostname has changed inside a group with same timestamps. Line " + lineNo);
						}
						data.values.put(type, valueD);
					}
					line = reader.readLine();
				}				
			}
		} catch (IOException e) {
			throw new IllegalStateException("Error handling data", e);
		}
		return data;
	}

	@Override
	public void initialize(String testName, InputStream in) {
		this.testName = testName;
		reader = new BufferedReader(new InputStreamReader(in));
		lineNo = 0;
		data = null;
		nextData = null;
	}
}
